package cn.doitedu.cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class Job6_Top10PayProducePerBrandV2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt/");
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 添加开启周期行提前触发的 窗口触发器
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled",true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay","1000 ms");


        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        "      id BIGINT," +
                        "      status INT,                        " +
                        "      payment_time timestamp(3),         " +
                        "      modify_time timestamp(3),          " +
                        "      rt as modify_time        ,         " +
                        "      watermark for rt as rt - interval '0' second ,  " +
                        "     PRIMARY KEY (id) NOT ENFORCED       " +
                        "     ) WITH (                            " +
                        "     'connector' = 'mysql-cdc',          " +
                        "     'hostname' = 'doitedu'   ,          " +
                        "     'port' = '3306'          ,          " +
                        "     'username' = 'root'      ,          " +
                        "     'password' = 'root'      ,          " +
                        "     'database-name' = 'realtimedw',           " +
                        "     'table-name' = 'oms_order'          " +
                        ")"
        );

        //tenv.executeSql("select * from order_mysql").print();
        /*
           +----+----+-------+-------------------------+-------------------------+-------------------------+
           | op | id |status |            payment_time |             modify_time |                      rt |
           +----+----+-------+-------------------------+-------------------------+-------------------------+
           | +I |  1 |     1 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 |
           | +I |  2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 |
           | +I |  3 |     1 | 2023-09-02 17:13:20.000 | 2023-09-02 17:13:20.000 | 2023-09-02 17:13:20.000 |
           | +I |  4 |     0 | 2023-09-02 17:20:20.000 | 2023-09-02 17:20:20.000 | 2023-09-02 17:20:20.000 |
       */


        tenv.executeSql(
                "CREATE TABLE item_mysql (    " +
                        "      id BIGINT," +
                        "      order_id BIGINT,                   " +
                        "      product_id BIGINT,                 " +
                        "      product_brand STRING,              " +
                        "      real_amount decimal(10,2),         " +
                        "     PRIMARY KEY (id) NOT ENFORCED       " +
                        "     ) WITH (                            " +
                        "     'connector' = 'mysql-cdc',          " +
                        "     'hostname' = 'doitedu'   ,          " +
                        "     'port' = '3306'          ,          " +
                        "     'username' = 'root'      ,          " +
                        "     'password' = 'root'      ,          " +
                        "     'database-name' = 'realtimedw',     " +
                        "     'table-name' = 'oms_order_item'     " +
                        ")"
        );

        //tenv.executeSql("select * from item_mysql").print();
        /*
            +----+----+---------+-----------+--------------+--------------+
            | op | id |order_id |product_id |product_brand |  real_amount |
            +----+----+---------+-----------+--------------+--------------+
            | +I |  1 |       1 |        20 |         华为 |      3585.98 |
            | +I |  2 |       1 |        21 |         华为 |      2022.81 |
            | +I |  3 |       1 |        22 |         小米 |       591.05 |
            | +I |  4 |       1 |        23 |         小米 |       636.23 |
            | +I |  5 |       1 |        26 |        apple |       636.23 |
            | +I |  6 |       1 |        27 |        apple |       636.23 |
            | +I |  7 |       1 |        28 |        apple |       636.23 |
            | +I |  8 |       2 |        21 |         华为 |      5496.06 |
            | +I |  9 |       2 |        22 |         小米 |      3585.98 |
            | +I | 10 |       2 |        23 |         小米 |      2022.81 |
            | +I | 11 |       2 |        24 |         华为 |       591.05 |
            | +I | 12 |       2 |        25 |         小米 |       591.05 |
            | +I | 13 |       2 |        26 |        apple |       591.05 |
            | +I | 14 |       2 |        27 |        apple |       591.05 |
            | +I | 15 |       2 |        28 |        apple |       591.05 |
            | +I | 16 |       3 |        23 |         小米 |       591.05 |
            | +I | 17 |       3 |        24 |         华为 |       591.05 |
            | +I | 18 |       3 |        25 |         小米 |       591.05 |
            | +I | 19 |       3 |        27 |        apple |       591.05 |
            | +I | 20 |       3 |        28 |        apple |       591.05 |
         */


        // 订单主表 和 订单商品项 关联
        tenv.executeSql(
                "create TEMPORARY view joined as           "+
                        " with od as (                              "+
                        " select *                                  "+
                        " from                                      "+
                        " order_mysql                               "+
                        " where                                     "+
                        "   payment_time is not null                "+
                        "   and status in (1,2,3)                   "+
                        " )                                         "+
                        " SELECT                                    "+
                        "   od.id AS oid,                           "+
                        "   od.status,                              "+
                        "   od.payment_time,                        "+
                        "   od.modify_time,                         "+
                        "   it.*                                    "+
                        " FROM  od                                  "+
                        " JOIN                                      "+
                        " item_mysql it                             "+
                        " ON od.id = it.order_id                    "
        );

        //tenv.executeSql("select * from joined").print();
        /*
            +----+-----+-------+-------------------------+-------------------------+----+---------+-----------+---------------+--------------+
            | op | oid |status |            payment_time |             modify_time | id |order_id |product_id | product_brand |  real_amount |
            +----+-----+-------+-------------------------+-------------------------+----+---------+-----------+---------------+--------------+
            | +I |   1 |     1 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 |  1 |       1 |        20 |          华为 |      3585.98 |
            | +I |   1 |     1 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 |  2 |       1 |        21 |          华为 |      2022.81 |
            | +I |   1 |     1 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 |  3 |       1 |        22 |          小米 |       591.05 |
            | +I |   1 |     1 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 |  4 |       1 |        23 |          小米 |       636.23 |
            | +I |   1 |     1 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 |  5 |       1 |        26 |         apple |       636.23 |
            | +I |   1 |     1 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 |  6 |       1 |        27 |         apple |       636.23 |
            | +I |   1 |     1 | 2023-09-02 16:10:20.000 | 2023-09-02 16:10:20.000 |  7 |       1 |        28 |         apple |       636.23 |
            | +I |   2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 |  8 |       2 |        21 |          华为 |      5496.06 |
            | +I |   2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 |  9 |       2 |        22 |          小米 |      3585.98 |
            | +I |   2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 | 10 |       2 |        23 |          小米 |      2022.81 |
            | +I |   2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 | 11 |       2 |        24 |          华为 |       591.05 |
            | +I |   2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 | 12 |       2 |        25 |          小米 |       591.05 |
            | +I |   2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 | 13 |       2 |        26 |         apple |       591.05 |
            | +I |   2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 | 14 |       2 |        27 |         apple |       591.05 |
            | +I |   2 |     1 | 2023-09-02 16:12:20.000 | 2023-09-02 16:12:20.000 | 15 |       2 |        28 |         apple |       591.05 |
            | +I |   3 |     1 | 2023-09-02 17:13:20.000 | 2023-09-02 17:13:20.000 | 16 |       3 |        23 |          小米 |       591.05 |
            | +I |   3 |     1 | 2023-09-02 17:13:20.000 | 2023-09-02 17:13:20.000 | 17 |       3 |        24 |          华为 |       591.05 |
            | +I |   3 |     1 | 2023-09-02 17:13:20.000 | 2023-09-02 17:13:20.000 | 18 |       3 |        25 |          小米 |       591.05 |
            | +I |   3 |     1 | 2023-09-02 17:13:20.000 | 2023-09-02 17:13:20.000 | 19 |       3 |        27 |         apple |       591.05 |
            | +I |   3 |     1 | 2023-09-02 17:13:20.000 | 2023-09-02 17:13:20.000 | 20 |       3 |        28 |         apple |       591.05 |
       */


        /**
         * 对关联后的数据
         * 通过表->流，再 流-> 表，重新定义rowtime
         */
        Schema schema = Schema.newBuilder()
                .column("order_id", DataTypes.BIGINT())
                .column("status", DataTypes.INT())
                .column("payment_time", DataTypes.TIMESTAMP(3))
                //.column("id", DataTypes.BIGINT())  // 详情表中的自增id
                .column("product_id", DataTypes.BIGINT())
                .column("real_amount", DataTypes.DECIMAL(10,2))
                .column("product_brand", DataTypes.STRING())
                .columnByExpression("rt", "modify_time")
                .watermark("rt", "rt - interval '0' second")
                .build();
        DataStream<Row> ds = tenv.toChangelogStream(tenv.from("joined"));
        // 流转表
        Table table = tenv.fromChangelogStream(ds, schema);

        tenv.createTemporaryView("timed",table);

        /**
         * 利用 groupByWindow  和 row_number  计算 窗口内各品牌成交额topn商品
         */
        tenv.executeSql(
                " create temporary view res as                                                    "+
                        "  with tmp as (                                                                   "+
                        "    SELECT                                                                        "+
                        "        tumble_start(rt,interval '24' hour) as w_start                           "+
                        "        ,tumble_end(rt,interval '24' hour) as w_end                              "+
                        "        ,tumble_rowtime(rt,interval '24' hour) as rt                             "+
                        "        ,product_brand                                                            "+
                        "        ,product_id                                                               "+
                        "        ,sum(real_amount) as real_amount                                          "+
                        "    from timed                                                                    "+
                        "    group by                                                                      "+
                        "    tumble(rt,interval '24' hour)                                                 "+
                        "    ,product_brand                                                                "+
                        "    ,product_id                                                                   "+
                        "  )                                                                               "+
                        "                                                                                  "+
                        " SELECT * FROM (                                                                  "+
                        " SELECT                                                                           "+
                        "   w_start,                                                                       "+
                        "   w_end,                                                                         "+
                        "   product_brand,                                                                 "+
                        "   product_id,                                                                    "+
                        "   real_amount,                                                                   "+
                        "   row_number() over(partition by w_start,w_end,product_brand order by real_amount desc) as rn  "+
                        " from tmp ) WHERE rn<=2                                                         "
        );

        //tenv.executeSql("select * from res").print();

        /*
            +----+-------------------------+-------------------------+---------------+------------+--------------+-----+
             | op |                 w_start |                   w_end | product_brand | product_id |  real_amount |  rn |
             +----+-------------------------+-------------------------+---------------+------------+--------------+-----+
             | +I | 2023-09-02 16:10:00.000 | 2023-09-02 16:11:00.000 |          华为 |         20 |      3585.98 |   1 |
             | +I | 2023-09-02 16:10:00.000 | 2023-09-02 16:11:00.000 |          华为 |         21 |      2022.81 |   2 |
             | +I | 2023-09-02 16:10:00.000 | 2023-09-02 16:11:00.000 |         apple |         28 |       636.23 |   1 |
             | +I | 2023-09-02 16:10:00.000 | 2023-09-02 16:11:00.000 |         apple |         27 |       636.23 |   2 |
             | +I | 2023-09-02 16:10:00.000 | 2023-09-02 16:11:00.000 |          小米 |         23 |       636.23 |   1 |
             | +I | 2023-09-02 16:10:00.000 | 2023-09-02 16:11:00.000 |          小米 |         22 |       591.05 |   2 |
             | +I | 2023-09-02 16:12:00.000 | 2023-09-02 16:13:00.000 |          华为 |         21 |      5496.06 |   1 |
             | +I | 2023-09-02 16:12:00.000 | 2023-09-02 16:13:00.000 |          小米 |         22 |      3585.98 |   1 |
             | +I | 2023-09-02 16:12:00.000 | 2023-09-02 16:13:00.000 |          华为 |         24 |       591.05 |   2 |
             | +I | 2023-09-02 16:12:00.000 | 2023-09-02 16:13:00.000 |         apple |         27 |       591.05 |   1 |
             | +I | 2023-09-02 16:12:00.000 | 2023-09-02 16:13:00.000 |         apple |         26 |       591.05 |   2 |
             | +I | 2023-09-02 16:12:00.000 | 2023-09-02 16:13:00.000 |          小米 |         25 |       591.05 |   2 |
             | -U | 2023-09-02 16:12:00.000 | 2023-09-02 16:13:00.000 |          小米 |         25 |       591.05 |   2 |
             | +U | 2023-09-02 16:12:00.000 | 2023-09-02 16:13:00.000 |          小米 |         23 |      2022.81 |   2 |
         */


        /* *
         *  创建mysql中的结果表映射 ------------------------------------------
         */
        tenv.executeSql(
                " CREATE TABLE mysql_sink (                      "
                        +"   product_brand     STRING,                       "
                        +"   rn   BIGINT,                                    "
                        +"   product_id   BIGINT,                            "
                        +"   real_amount   DECIMAL(10,2),             "
                        +"   PRIMARY KEY (product_brand,rn) NOT ENFORCED "
                        +" ) WITH (                                          "
                        +"    'connector' = 'jdbc',                          "
                        +"    'url' = 'jdbc:mysql://doitedu:3306/rtmk',      "
                        +"    'table-name' = 'topn_amount_brand',            "
                        +"    'username' = 'root',                           "
                        +"    'password' = 'root'                            "
                        +" )                                                 "
        );

        /**
         * 输出最终结果
         */
        tenv.executeSql("insert into mysql_sink select product_brand,rn,product_id,real_amount from  res");


        env.execute();
    }

}